package com.fowo.api.sys.acm;

import com.baomidou.mybatisplus.core.incrementer.DefaultIdentifierGenerator;
import com.baomidou.mybatisplus.core.incrementer.IdentifierGenerator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fowo.api.common.model.R;
import com.fowo.api.user.model.JwtUserInfo;
import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 异步控制器方法执行者
 * @author yl_ls
 */
@Slf4j
@Component
public class AsyncControllerMethodExecutor implements ApplicationContextAware {
    /**
     * 最大等待任务数
     */
    public static final int MAX_QUEUE_SIZE = 5;
    /**
     * 已完成任务在缓存中最大等待被读取的时间，到时后会被清除结果
     */
    public static final int MAX_FINISH_CACHE_MILLIS = 60000;
    private final IdentifierGenerator identifierGenerator = new DefaultIdentifierGenerator();
    private final Thread thread = new Thread(this::run);
    private final BlockingQueue<TaskInfo> taskInfoQueue = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);
    private final Map<Long, TaskInfo> finishTaskMaps = new HashMap<>();
    private boolean hasShutdown = false;

    public AsyncControllerMethodExecutor() {
        Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
    }

    public Object startTask(ProceedingJoinPoint joinPoint) {
        if (taskInfoQueue.size() == MAX_QUEUE_SIZE) {
            return R.fail("当前未完成长任务过多，请稍后再试");
        }
        Long newTaskId = (Long) identifierGenerator.nextId(null);
        taskInfoQueue.add(new TaskInfo().taskId(newTaskId).joinPoint(joinPoint));
        return R.data(newTaskId);
    }

    public TaskInfo getResult(Long taskId, boolean removeIfFinish) {
        TaskInfo taskInfo = finishTaskMaps.get(taskId);
        if (taskInfo != null) {
            if (removeIfFinish) {
                finishTaskMaps.remove(taskId);
            }
            return taskInfo;
        }
        return taskInfoQueue.stream().filter(ti -> Objects.equals(taskId, ti.taskId)).findFirst().orElseGet(() -> {
            // 没有在完成中，也不在待运行中，任务已不存在，返回未执行错误
            return new TaskInfo()
                    .taskId(taskId)
                    .finish(true)
                    .result(R.fail("此任务未能执行，请重试"));
        });
    }

    public void run() {
        log.debug("长任务线程已启动");
        while (!hasShutdown) {
            try {
                TaskInfo taskInfo = taskInfoQueue.peek();
                if (taskInfo == null) {
                    clearFinishCache();
                    Thread.sleep(10);
                    continue;
                }
                if (taskInfo.cancelled) {
                    taskInfoQueue.poll();
                    log.debug("长任务 {} 已取消，跳过处理", taskInfo.taskId);
                    continue;
                }
                runTask(taskInfo);
                finishTaskMaps.put(taskInfo.taskId, taskInfo);
                taskInfoQueue.poll();
            } catch (Exception e) {
                log.error("处理长任务时发生未处理错误！");
            }
        }
        log.debug("长任务线程已终止");
    }

    private void clearFinishCache() {
        if (finishTaskMaps.isEmpty()) {
            return;
        }
        long current = System.currentTimeMillis();
        List<Long> removeKeys = new ArrayList<>();
        for (Map.Entry<Long, TaskInfo> longTaskInfoEntry : finishTaskMaps.entrySet()) {
            if (longTaskInfoEntry.getValue().cacheTime < current) {
                removeKeys.add(longTaskInfoEntry.getKey());
            }
        }
        removeKeys.forEach(finishTaskMaps::remove);
    }

    private void runTask(TaskInfo taskInfo) {
        log.debug("开始执行长任务{}", taskInfo.taskId);
        if (taskInfo.user != null) {
            JwtUserInfo.setCurrent(taskInfo.user);
        }
        try {
            taskInfo.running(true);
            taskInfo.result(taskInfo.joinPoint.proceed());
        } catch (Throwable e) {
            taskInfo.result(R.fail(e.getMessage()));
        }
        JwtUserInfo.setCurrent(null);
        taskInfo.running(false);
        taskInfo.finish(true);
        taskInfo.cacheTime(System.currentTimeMillis() + MAX_FINISH_CACHE_MILLIS);
        taskInfo.joinPoint = null;
    }

    public void shutdown() {
        hasShutdown = true;
        try {
            thread.join(10000);
        } catch (InterruptedException e) {
            thread.interrupt();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        thread.setName("AsyncControllerMethodExecutor");
        thread.start();
    }

    @Getter
    @Setter
    @Accessors(chain = true, fluent = true)
    static class TaskInfo {
        private Long taskId;
        private JwtUserInfo user;
        @JsonIgnore
        private transient ProceedingJoinPoint joinPoint;
        private boolean cancelled;
        private boolean running;
        private boolean finish;
        private long cacheTime;
        private Object result;
    }
}
